package com.lnj.messages.consumer;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.IndexRequest;
import co.elastic.clients.elasticsearch.core.IndexResponse;
import com.lnj.bean.dto.ProductDTO;
import com.lnj.bean.dto.UserDTO;
import com.lnj.messages.servise.RecommenderService;
import com.lnj.messages.utiles.JsonUtil;
import com.lnj.messages.utiles.MailClientUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Service
@Slf4j
public class KafkaConsumer {
    @Autowired
    private ElasticsearchClient elasticsearchClient;
    @Autowired
    private MailClientUtil mailClientUtil;
    @Autowired
    private JsonUtil jsonUtil;
    @Autowired
    private RecommenderService recommenderService;


    @KafkaListener(topics = "es_data_topic")
    public void handleEsData(ConsumerRecord<String, String> record)  {
        log.info("es_data_topic消费消息: {}", record.value());
        String key = record.key();
        String value = record.value();

        try {
            if ("products".equals(key)) {
                // 处理商品数据
                ProductDTO productDTO = jsonUtil.fromJson(value, ProductDTO.class);
                IndexRequest<ProductDTO> request = buildIndexRequest(productDTO,productDTO.getProductId(), "products");
                IndexResponse response = elasticsearchClient.index(request);
            }else if ("users".equals(key)) {
                // 处理用户数据
                UserDTO userDTO = jsonUtil.fromJson(value, UserDTO.class);
                IndexRequest<UserDTO> request = buildIndexRequest(userDTO,userDTO.getUserId(), "users");
                IndexResponse response = elasticsearchClient.index(request);
            }
        } catch (Exception e) {
            log.error("处理消息ES数据失败: {}",record.key(),record.value());
        }
    }


    @KafkaListener(topics = "email_topic")
    public void handleEmailData(ConsumerRecord<String, String> record) {
        String key = record.key();
        String value = record.value();
        log.info("email_topic消费消息: {}", record.value());
        if ("user_register".equals(key)) {
            // 处理用户注册数据
            UserDTO userDTO = jsonUtil.fromJson(value, UserDTO.class);
            // 注册时间
            Date date = new Date();
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String format = sdf.format(date);
            Map<String,Object>map=new HashMap<>();
            map.put("userName",userDTO.getUserName());
            // 注册时间
            map.put("registerTime",format);
            map.put("loginUrl","http://localhost:8080/login");
            mailClientUtil.sendTemplateMail(userDTO.getEmail(),"用户注册成功","userRegister",map);
        }
    }

    @KafkaListener(topics = "recommend_topic")
    public void handleRecommendData(String record) {
        Long userId = Long.valueOf(record);
        log.info("recommend_topic消费消息: {}", record);
        // 推荐商品
        recommenderService.recommendItems(userId);
    }


    private <T> IndexRequest<T> buildIndexRequest(T message,Long id, String indexName) {
        return IndexRequest.of(b -> b
                .index(indexName)
                .id(id+"")
                .routing(calculateRouting(id))
                .document(message)
        );
    }
    private String calculateRouting(long productId) {
        return "shard_" + (productId % 10); // 更确定性的分片路由
    }
}
